package com.tdy.cdc.app.dwd;

import com.tdy.cdc.app.BaseSqlApp;
import com.tdy.cdc.common.Constant;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.FlinkSourceUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author NanHuang
 * @Date 2023/3/14
 */
public class App07_DwdMhsGeneralSurgeryRecord extends BaseSqlApp {
    public static void main(String[] args) {
        new App07_DwdMhsGeneralSurgeryRecord().init(
                5007,
                "App7_DwdMhsGeneralSurgeryRecord",
                2
        );
    }

    @Override
    protected void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 创建源表
        createSourceTables(tableEnv);
        // 2 join
        joinTables(tableEnv);
        // 3 写入Kafka
        writeToKafka(tableEnv);
    }

    private void writeToKafka(StreamTableEnvironment tableEnv) {
        // 创建Kafka映射表
        tableEnv.executeSql("create table target (" +
                "RECORD_ID                     String,"+
                "MED_ORG_NAME                  String,"+
                "HOSPITAL_NAME                 String,"+
                "HOSPITAL_CODE                 String,"+
                "AGE_UNIT                      String,"+
                "AGE                           String,"+
                "SEX_NAME                      String,"+
                "SEX_CODE                      String,"+
                "ID_CARD_NUMBER                String,"+
                "PATIENT_VISIT_CATEGORY_CODE   String,"+
                "PATIENT_VISIT_CATEGORY        String,"+
                "INPATIENT_VISIT_FLOW_ID       String,"+
                "INPATIENT_NO                  String,"+
                "OUTPATIENT_VISIT_FLOW_ID      String,"+
                "OUTPATIENT_NO                 String,"+
                "PATIENT_ID                    String,"+
                "PATIENT_NAME                  String,"+
                "DOMAIN_ID                     String,"+
                "WARD_ID                       String,"+
                "WARD_NAME                     String,"+
                "VISIT_TIMES                   String,"+
                "APPLY_SHEET_NO                String,"+
                "AUTHOR_CODE                   String,"+
                "AUTHOR_NAME                   String,"+
                "CUSTODIAN_CODE                String,"+
                "CUSTODIAN_NAME                String,"+
                "SURGEON_SIGNDATE              String,"+
                "VISIT_DATE                    String,"+
                "BED_NO                        String,"+
                "BED_NAME                      String,"+
                "DEPT_ID                       String,"+
                "DEPT_NAME                     String,"+
                "SURGERY_HISTORY               String,"+
                "OPER_CODE                     String,"+
                "OPER_NAME                     String,"+
                "OPER_LEVEL_CODE               String,"+
                "OPER_LEVEL                    String,"+
                "OPER_START_DATE               String,"+
                "OPER_END_DATE                 String,"+
                "FIRST_ASSISTANT_ID            String,"+
                "FIRST_ASSISTANT_NAME          String,"+
                "SECOND_ASSISTANT_ID           String,"+
                "SECOND_ASSISTANT_NAME         String,"+
                "INSTRUMENT_NURSE_ID           String,"+
                "INSTRUMENT_NURSE_NAME         String,"+
                "VISITATORIAL_NURSE_ID         String,"+
                "VISITATORIAL_NURSE_NAME       String,"+
                "OPER_ROOM                     String,"+
                "BLEEDING                      String,"+
                "METACHYSIS_VOLUME             String,"+
                "METACHYSIS_REACTION_FLAG      String,"+
                "ANES_METHOD_CODE              String,"+
                "ANES_METHOD_NAME              String,"+
                "PRE_MEDICATION                String,"+
                "INOPER_MEDICAL                String,"+
                "INFUSION_VOLUME               String,"+
                "OPERATION_DESC                String,"+
                "TARGET_NAME                   String,"+
                "INTERVENTION                  String,"+
                "OPER_POST_CODE                String,"+
                "OPER_POST                     String,"+
                "SKIN_DISINFECTION             String,"+
                "INCISION_DESC                 String,"+
                "DRAINAGE_MARK                 String,"+
                "DRAINAGE_MATERIAL             String,"+
                "DRAINAGE_MATERIAL_NUM         String,"+
                "PLACEMENT_SITE                String,"+
                "ROOM_NAME                     String,"+
                "ANES_DOCTOR_NAME              String,"+
                "ROOM_NO                       String,"+
                "PRE_OPER_DIAG_ICD_NAME        String,"+
                "PRE_OPER_DIAG_ICD_CODE        String,"+
                "POST_OPER_DIAG_ICD_CODE       String,"+
                "POST_OPER_DIAG_ICD_NAME       String," +
                "VISIT_ID                      STRING,"+
                "primary key (RECORD_ID) not enforced" +
                ")" +
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_GENERAL_SURGERY_RECORD));
        // 导入数据
        tableEnv.executeSql("insert into target select * from join_result");
    }






    private void joinTables(StreamTableEnvironment tableEnv) {
        // 1 设置数据的TTL（暂时10min）
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60));
        // 2 join操作
        Table joinResult = tableEnv.sqlQuery("select     " +
                "s.unique_id                                                     AS   RECORD_ID   ,"+
                "hospital_name                                                   AS   MED_ORG_NAME   ,"+
                "hospital_name                                                   AS   HOSPITAL_NAME   ,"+
                "hospital_code                                                   AS   HOSPITAL_CODE   ,"+
                "age_unit                                                        AS   AGE_UNIT   ,"+
                "age                                                             AS   AGE   ,"+
                "sex                                                             AS   SEX_NAME   ,"+
                "sex_code                                                        AS   SEX_CODE   ,"+
                "identify_no                                                     AS   ID_CARD_NUMBER   ,"+
                "patient_typecode                                                AS   PATIENT_VISIT_CATEGORY_CODE   ,"+
                "patient_type                                                    AS   PATIENT_VISIT_CATEGORY   ,"+
                "case when patient_typecode ='3' then visit_id else '' end       AS   INPATIENT_VISIT_FLOW_ID   ,"+
                "inpatient                                                       AS   INPATIENT_NO   ,"+
                "case when patient_typecode <> '3' then visit_id else '' end      AS   OUTPATIENT_VISIT_FLOW_ID   ,"+
                "out_patient_id                                                  AS   OUTPATIENT_NO   ,"+
                "patient_id                                                      AS   PATIENT_ID   ,"+
                "patient_name                                                    AS   PATIENT_NAME   ,"+
                "s.domain_id                                                     AS   DOMAIN_ID   ,"+
                "wards_id                                                        AS   WARD_ID   ,"+
                "wards_name                                                      AS   WARD_NAME   ,"+
                "visit_times                                                     AS   VISIT_TIMES   ,"+
                "appy_id                                                         AS   APPLY_SHEET_NO   ,"+
                "author_code                                                     AS   AUTHOR_CODE   ,"+
                "author_name                                                     AS   AUTHOR_NAME   ,"+
                "custodian_code                                                  AS   CUSTODIAN_CODE   ,"+
                "custodian_name                                                  AS   CUSTODIAN_NAME   ,"+
                "surgeon_signdate                                                AS   SURGEON_SIGNDATE   ,"+
                "visit_date                                                      AS   VISIT_DATE   ,"+
                "bed_no                                                          AS   BED_NO   ,"+
                "bed_name                                                        AS   BED_NAME   ,"+
                "dept_code                                                       AS   DEPT_ID   ,"+
                "dept_name                                                       AS   DEPT_NAME   ,"+
                "surgery_history                                                 AS   SURGERY_HISTORY   ,"+
                "oper_code                                                       AS   OPER_CODE   ,"+
                "oper_name                                                       AS   OPER_NAME   ,"+
                "oper_level_code                                                 AS   OPER_LEVEL_CODE   ,"+
                "oper_level                                                      AS   OPER_LEVEL   ,"+
                "oper_start_date                                                 AS   OPER_START_DATE   ,"+
                "oper_end_date                                                   AS   OPER_END_DATE   ,"+
                "assistant_icode                                                 AS   FIRST_ASSISTANT_ID   ,"+
                "assistant_i                                                     AS   FIRST_ASSISTANT_NAME   ,"+
                "assistant_iicode                                                AS   SECOND_ASSISTANT_ID   ,"+
                "assistant_ii                                                    AS   SECOND_ASSISTANT_NAME   ,"+
                "instrument_code                                                 AS   INSTRUMENT_NURSE_ID   ,"+
                "instrument_name                                                 AS   INSTRUMENT_NURSE_NAME   ,"+
                "patrol_code                                                     AS   VISITATORIAL_NURSE_ID   ,"+
                "patrol_name                                                     AS   VISITATORIAL_NURSE_NAME   ,"+
                "oper_room                                                       AS   OPER_ROOM   ,"+
                "bleeding                                                        AS   BLEEDING   ,"+
                "blood_volume                                                    AS   METACHYSIS_VOLUME   ,"+
                "cast(case when blood_reaction_mark='1' then blood_reaction_mark='true' else blood_reaction_mark='false' end as string)  AS   METACHYSIS_REACTION_FLAG   ,"+
                "anes_method_code                                                AS   ANES_METHOD_CODE   ,"+
                "anes_method_name                                                AS   ANES_METHOD_NAME   ,"+
                "pre_medication                                                  AS   PRE_MEDICATION   ,"+
                "inoper_medical                                                  AS   INOPER_MEDICAL   ,"+
                "infusion_volume                                                 AS   INFUSION_VOLUME   ,"+
                "operation_desc                                                  AS   OPERATION_DESC   ,"+
                "target_name                                                     AS   TARGET_NAME   ,"+
                "intervention                                                    AS   INTERVENTION   ,"+
                "oper_post_code                                                  AS   OPER_POST_CODE   ,"+
                "oper_post                                                       AS   OPER_POST   ,"+
                "skin_disinfection                                               AS   SKIN_DISINFECTION   ,"+
                "incision_desc                                                   AS   INCISION_DESC   ,"+
                "cast(case when drainage_mark ='1' then drainage_mark ='true' else drainage_mark ='false' end as string)  AS   DRAINAGE_MARK   ,"+
                "drainage_material                                               AS   DRAINAGE_MATERIAL   ,"+
                "drainage_material_num                                           AS   DRAINAGE_MATERIAL_NUM   ,"+
                "placement_site                                                  AS   PLACEMENT_SITE   ,"+
                "ward_name                                                       AS   ROOM_NAME      ,"+
                "anesth_name                                                     AS   ANES_DOCTOR_NAME      ,"+
                "ward_id                                                         AS   ROOM_NO      ,"+
                "case when diag_type ='术前诊断' then diag_code else '' end     AS   PRE_OPER_DIAG_ICD_CODE     ,"+
                "case when diag_type ='术前诊断' then diag_name else '' end       AS   PRE_OPER_DIAG_ICD_NAME     ,"+
                "case when diag_type ='术后诊断' then diag_code else '' end       AS   POST_OPER_DIAG_ICD_CODE     ,"+
                "case when diag_type ='术后诊断' then diag_name else '' end       AS   POST_OPER_DIAG_ICD_NAME ," +
                "visit_id                                                        AS   VISIT_ID "+
                "from hdsd00_06_06 s " +
                "left join tdy_list_diag a " +
                "on a.unique_id=s.unique_id and a.xds_type=s.xds_type"
        );
        tableEnv.createTemporaryView("join_result",joinResult);
    }




    private void createSourceTables(StreamTableEnvironment tableEnv) {
        // hdsd00_06_06
        tableEnv.executeSql("create table hdsd00_06_06 (" +
                "pk                          String,"+
                "upload_time                 String,"+
                "status                      String,"+
                "empi                        String,"+
                "encounter_id                String,"+
                "bed_name                    String,"+
                "ward_name                   String,"+
                "wards_id                    String,"+
                "patient_id                  String,"+
                "patient_domain              String,"+
                "visit_domain                String,"+
                "visit_id                    String,"+
                "visit_times                 String,"+
                "unique_id                   String,"+
                "xds_type                    String,"+
                "xds_name                    String,"+
                "effective_time              String,"+
                "xds_version                 String,"+
                "domain_id                   String,"+
                "patient_type                String,"+
                "out_patient_id              String,"+
                "patient_typecode            String,"+
                "inpatient                   String,"+
                "appy_id                     String,"+
                "identify_no                 String,"+
                "patient_name                String,"+
                "sex_code                    String,"+
                "sex                         String,"+
                "age                         String,"+
                "age_unit                    String,"+
                "write_time                  String,"+
                "author_code                 String,"+
                "author_name                 String,"+
                "custodian_code              String,"+
                "custodian_name              String,"+
                "surgeon_signdate            String,"+
                "visit_date                  String,"+
                "bed_no                      String,"+
                "ward_id                     String,"+
                "dept_code                   String,"+
                "dept_name                   String,"+
                "wards_name                  String,"+
                "hospital_code               String,"+
                "hospital_name               String,"+
                "surgery_history             String,"+
                "oper_code                   String,"+
                "oper_start_date             String,"+
                "oper_end_date               String,"+
                "surgeon_code                String,"+
                "surgeon_name                String,"+
                "assistant_icode             String,"+
                "assistant_i                 String,"+
                "assistant_iicode            String,"+
                "assistant_ii                String,"+
                "instrument_code             String,"+
                "instrument_name             String,"+
                "patrol_code                 String,"+
                "patrol_name                 String,"+
                "oper_name                   String,"+
                "oper_room                   String,"+
                "oper_level_code             String,"+
                "oper_level                  String,"+
                "bleeding                    String,"+
                "blood_volume                String,"+
                "blood_reaction_mark         String,"+
                "anes_method_code            String,"+
                "anes_method_name            String,"+
                "anesth_name                 String,"+
                "pre_medication              String,"+
                "inoper_medical              String,"+
                "infusion_volume             String,"+
                "operation_desc              String,"+
                "target_name                 String,"+
                "intervention                String,"+
                "oper_post_code              String,"+
                "oper_post                   String,"+
                "skin_disinfection           String,"+
                "incision_desc               String,"+
                "drainage_mark               String,"+
                "drainage_material           String,"+
                "drainage_material_num       String,"+
                "placement_site              String"+
                ")" +
                FlinkSourceUtil.getKafkaWith("hdsd00_06_06","App07_DwdMhsGeneralSurgeryRecord"));


        // tdy_list_diag
        super.readTdyListDiag(tableEnv,"App07_DwdMhsGeneralSurgeryRecord");
    }
}
